package com.ebs.md.fxmonitor.client.impl;

import java.net.InetAddress;
import java.net.Socket;
import java.util.Arrays;
import java.util.Map;
import java.util.Set;

import org.apache.log4j.Logger;
import com.cameronsystems.fix.Session;
import com.cameronsystems.fix.configuration.Constants;
import com.cameronsystems.fix.connection.Connection;
import com.cameronsystems.fix.connection.SocketConnection;
import com.cameronsystems.fix.exception.MessageException;
import com.cameronsystems.fix.inprocess.InProcessAdapter;
import com.cameronsystems.fix.inprocess.InProcessAdapterProcessor;
import com.cameronsystems.fix.inprocess.InProcessEventListener;
import com.cameronsystems.fix.manager.ISessionManager;
import com.cameronsystems.fix.message.InProcessEvent;
import com.cameronsystems.fix.message.Message;
import com.ebs.commons.translators.fix_translator.api.ICAPFixConstants;
import com.ebs.commons.comp.IParsedPayload;
import com.ebs.commons.comp.ParsedPayload;
import com.ebs.commons.logger.MessageText;
import com.ebs.md.fxmonitor.client.api.ClientMessageListener;
import com.ebs.md.fxmonitor.client.api.FixMessageSender;
import com.ebs.md.fxmonitor.topics.ClientTopicKeys;
import com.orcsoftware.util.data.ConversionException;

/**
 * <p>Title: Netlink</p>
 *
 * <p>Copyright: Copyright (c) 2004</p>
 *
 * <p>Company: EBS</p>
 * 
 * Revision Author:
 *  $File: //depot/MarketData/fxmonitor/main/src/main/java/com/ebs/md/fxmonitor/client/impl/FixInProcessSession.java $
 *  $Revision: #7 $
 *  $DateTime: 2012/07/29 11:38:46 $
 *  $Change: 112707 $
 *  $Author: Arik.Sher $
 */
public class FixInProcessSession extends InProcessEventListener implements FixMessageSender {
    static final public String APPLICATION_ID = "FXMonServer";
    static final public String PARTY_ID = "FXMonClient";
    static final public String CONNECT_STATUS = "connected";
    static final public String DISCONNECT_STATUS = "disconnected";

    private InProcessAdapter ipa;
    private ISessionManager m_sessionManager;
    private String m_clientIp;

    private ClientMessageListener m_clientMessageListener;
    private String m_cameronConfigFile;

    final static Logger s_logger = Logger.getLogger(FixInProcessSession.class);

    public FixInProcessSession() {
        super();
        m_clientIp = "unknown";
    }

    public void setClientMessageListener(ClientMessageListener list) {
        m_clientMessageListener = list;
    }

    private Session getClientSession() {
        return m_sessionManager.getSession();
    }

    public String getClientIp() {
        return m_clientIp;
    }

    private void setClientIp() {
        m_clientIp = "unknown";

        Session sess = getClientSession();
        if (sess != null) {
            Connection conn = sess.getConnection();
            if (conn != null && conn instanceof SocketConnection) {
                Socket socket = ( (SocketConnection) conn).getSocket();
                InetAddress address = socket.getInetAddress();
                m_clientIp = address.toString();
            }
        }
    }

    public void setCameronConfig(java.io.File config) {
        m_cameronConfigFile = config.getPath();
    }

    public void init() throws Exception {
    }

    public void start() throws Exception {
        if (null == m_cameronConfigFile) {
            throw new RuntimeException("Attribute \"cameronConfig\" is not set.");
        }
        s_logger.info(new MessageText("FXMON_0500019"));//"starting adapter..."
        String[] args = {"-xmlconfig", m_cameronConfigFile, "-id", APPLICATION_ID, "-nogui"};
        ipa = new InProcessAdapter(args);
        ipa.addListener(PARTY_ID, this);

        Map processors = ipa.getProcessors();
        Set keySet = processors.keySet();
//        final int size = keySet.size();

        InProcessAdapterProcessor ipap = ipa.getProcessor(PARTY_ID);
        m_sessionManager = ipap.getSessionManager();

        s_logger.info(new MessageText("FXMON_0500020", Arrays.toString(keySet.toArray())));//"adapter has started"
        m_clientMessageListener.onServerInitialized();

    }


    public void onEvent(Object evt) {
        InProcessEvent event = (InProcessEvent) evt;
        try {

            if (event.getType() == InProcessEvent.TYPE_FIX_MESSAGE) {
                Message msg = (Message) event.getEventData();
                if(s_logger.isDebugEnabled()){
                    s_logger.debug(new MessageText("FXMON_0500016", msg.toString(1)));//"incoming message:"
                }
                if (Constants.MSGLogon.equals(msg.getMsgType())) {
                    onClientLogin(msg);
//                    onClientSubscribe(msg);//we don't have an explicit subscription in FX Monitor
                }
                else if (Constants.MSGLogout.equals(msg.getMsgType())) {
                    onClientLogout(msg);
                }
                else if (msg.getMsgType().equalsIgnoreCase(ICAPFixConstants.TypeCustomRequest)) {
                    String subtype = msg.getStringFieldValue(ICAPFixConstants.TAGSubtype);
                    if (subtype.equalsIgnoreCase(ICAPFixConstants.SubtypeSubscription)) {
                    	//Currently we don't receive subscription requests. I.N. 07/23/2008
//                        onClientSubscribe(msg);
                    }
                    else {
                        s_logger.error(new MessageText("FXMON_0500015", msg.toString(1)));//"Unknown message:"
                    }
                }
            }
            else if (event.getType() == InProcessEvent.TYPE_CONNECTION_STATUS) {
                String status = (String) event.getEventData();
                if (CONNECT_STATUS.equalsIgnoreCase(status)) {
                    onClientConnect();
                }
                else if (DISCONNECT_STATUS.equalsIgnoreCase(status)) {
                    onClientDisconnect();
                }
            }
            else if (event.getType() == InProcessEvent.TYPE_RESET) {
                String status = (String) event.getEventData();
                String session = event.getSource();
                s_logger.warn(new MessageText("FXMON_0500017", session, status));//"reset
            }
            else if (event.getType() == InProcessEvent.TYPE_EXCEPTION) {
                Object ex = event.getEventData();
                onClientException( (Exception) ex);
            }
            else {
                s_logger.error(new MessageText("FXMON_0500018", evt));//"Unknown event:"
            }
        }
        catch (MessageException me) {
            MessageText mt = new MessageText("FXMON_0100004");
            s_logger.error(mt, me);
            onProtocolFailure(mt.getText());
            onClientException(me);
        }
        catch (Exception me) {
            MessageText mt = new MessageText("FXMON_0100004");
            s_logger.error(mt, me);
            onProtocolFailure(mt.getText());
            onClientException(me);
        }
    }

    private void onClientSubscribe(Message message) throws MessageException {
        m_clientMessageListener.onClientSubscribe(convertSubscribeMessage(message));
    }

    private void onClientConnect() {
        setClientIp();
        m_clientMessageListener.onClientConnected(getClientIp());
    }

    private void onClientDisconnect() {
        m_clientMessageListener.onClientDisconnected("disconnect");
//        System.out.println("----disconnected " + m_sessionManager.getSession().toString());
    }

    private void onClientLogin(Message message) throws MessageException, ConversionException {
        m_clientMessageListener.onClientLogin(convertLoginMessage(message));
    }

    private void onClientLogout(Message message) {
        m_clientMessageListener.onClientLogout("Gracefull logout", true);
    }

    private void onClientException(Exception ex) {
        m_clientMessageListener.onClientException(ex);
    }

    public void send(Message message) {
        try {
            ipa.send(PARTY_ID, message);
        }
        catch (Exception ex) {
            onClientException(ex);
        }
    }

    public void closeSession() {
        m_sessionManager.dropSession();
    }

    private IParsedPayload convertLoginMessage(Message message) throws MessageException, ConversionException {
        IParsedPayload payload = new ParsedPayload();
        //parse the Fix message first
        payload.addField(ClientTopicKeys.ClientProtocolVersion, message.getValueAsString(ICAPFixConstants.TAGiClientProtocolVersion, "1"));

        boolean resetFlag = message.getValueAsBoolean(141, false);//Integer.getInteger(Constants.TAGResetSeqNumFlag));
        int flag = resetFlag ? 1 : 0;
        payload.addField(ClientTopicKeys.ResetSequenceNumberFlag, flag);

        //add auxillary
        payload.addField(ClientTopicKeys.ClientIp, getClientIp());
        payload.addField(ClientTopicKeys.ClientName, PARTY_ID);

        return payload;
    }

    private static IParsedPayload convertSubscribeMessage(Message message) throws MessageException{
        IParsedPayload payload = new ParsedPayload();
        payload.addField(ClientTopicKeys.ArbMessageTime, 0);
        payload.addField(ClientTopicKeys.ArbSequenceNum, 0);
        payload.addField(ClientTopicKeys.ArbSessionNum, 0);
        return payload;
    }

    private void onProtocolFailure(String reason){
        m_clientMessageListener.onClientLogout(reason, false);//forced logout - don't need to notify the listener if logout request came from the mediator itself
        logoutClient(reason);
    }

    public void logoutClient(String reason) {
        try {
            m_sessionManager.connectSession(false, reason);
        }
        catch (Exception ex) {
            ex.printStackTrace();
        }
    }
}
